package com.lanjiu.lib.communication;

import com.lanjiu.pro.business.BusinessProtocolMessageStandard;
import com.lanjiu.lib.communication.callback.OnConnectStatusCallback;
import com.lanjiu.lib.communication.dispatcher.MsgDispatcher;
import com.lanjiu.lib.communication.event.OnEventListener;
import com.lanjiu.lib.communication.interf.IMSClientInterface;
import com.lanjiu.lib.communication.netty.TCPChannelInitializer;
import com.lanjiu.lib.communication.netty.TCPReadHandler;
import com.lanjiu.lib.communication.timer.MsgTimeoutTimerManager;
import com.lanjiu.lib.communication.timer.RequestTimeoutTimerManager;
import com.lanjiu.lib.communication.work.ExecutorServiceFactory;
import com.lanjiu.lib.utils.log.Logger;
import com.lanjiu.lib.utils.th.Threader;

import java.net.InetSocketAddress;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.StringUtil;

/**
 * @description: 基于netty实现的tcp ims
 * @author: Panwei
 * @date: 2019/10/28 15:29
 */
public class NettyTCPClient implements IMSClientInterface {

    private static volatile NettyTCPClient mInstance;

    static NettyTCPClient getInstance() {
        if (null == mInstance) {
            synchronized (NettyTCPClient.class) {
                if (null == mInstance) {
                    mInstance = new NettyTCPClient();
                }
            }
        }
        return mInstance;
    }

    private Bootstrap mBootstrap;
    private Channel mChannel;

    private boolean isClosed = false;   // 标识ims是否已关闭
    private String mServerUrl;      // 服务器地址  ip:port
    private OnEventListener mOnEventListener; // 与应用曾交互的listener
    private OnConnectStatusCallback mConnectStatusCallback; // ims连接状态回调
    private MsgDispatcher mMsgDispatcher;   // 消息转发器
    private ExecutorServiceFactory mResetConnectExecutor;  // 重连线程池

    private boolean isConnecting = false;// 是否正在进行连接
    private int mConnectStatus = IMSConfig.CONNECT_STATE_FAILURE;// ims连接状态，初始化为连接失败
    private String mCurrentHost = null;// 当前连接host
    private int mCurrentPort = -1;// 当前连接port
    private MsgTimeoutTimerManager mMsgTimeoutTimerManager;
    private RequestTimeoutTimerManager mRequestTimeoutTimerManager;
    private int mConnectFailureRetryCount;

    private NettyTCPClient() {
    }

    @Override
    public void init(String serverUrl, OnEventListener listener, OnConnectStatusCallback callback) {
        close();
        isClosed = false;
        mServerUrl = serverUrl;
        mOnEventListener = listener;
        mConnectStatusCallback = callback;
        mMsgDispatcher = new MsgDispatcher();
        mMsgDispatcher.setOnEventListener(listener);
        mResetConnectExecutor = new ExecutorServiceFactory();
        mResetConnectExecutor.initBossLoopGroup();
        mMsgTimeoutTimerManager = new MsgTimeoutTimerManager(this);
        mRequestTimeoutTimerManager = new RequestTimeoutTimerManager(this);

        // 进行第一次连接
        connect();
    }

    @Override
    public boolean isNetworkAvailable() {
        if (mOnEventListener != null)
            return mOnEventListener.isNetworkAvailable();
        return false;
    }

    @Override
    public void connect() {
        doConnect(true);
    }

    @Override
    public void resetConnect() {
        doConnect(false);
    }

    @Override
    public boolean isConnect() {
        if (mChannel != null)
            return mChannel.isActive();
        return false;
    }

    @Override
    public boolean close() {
        if (isClosed) {
            return true;
        }
        if (mMsgTimeoutTimerManager != null && mMsgTimeoutTimerManager.hasSending()) {
            Logger.print("有消息正在发送中不允许关闭连接");
            return false;
        }
        isClosed = true;

        // 关闭channel
        try {
            closeChannel();
        } catch (Exception ex) {
            ex.printStackTrace();
        }

        // 关闭bootstrap
        try {
            if (mBootstrap != null) {
                mBootstrap.group().shutdownGracefully();
            }
        } catch (Exception ex) {
            ex.printStackTrace();
        }

        if (mOnEventListener != null)
            mOnEventListener.close();

        try {
            // 释放线程池
            if (mResetConnectExecutor != null)
                mResetConnectExecutor.destroyBossLoopGroup();
        } finally {
            isConnecting = false;
            mChannel = null;
            mBootstrap = null;
        }
        return true;
    }

    @Override
    public boolean isClosed() {
        return isClosed;
    }

    @Override
    public void sendMsg(BusinessProtocolMessageStandard.CheckUnifiedEntranceMessage msg) {
        sendMsg(msg, true);
    }

    @Override
    public void sendMsg(BusinessProtocolMessageStandard.CheckUnifiedEntranceMessage msg, boolean isJoinTimeoutManager) {
        if (msg == null) {
            Logger.print("发送消息失败，消息为空");
            return;
        }
        if (!isNetworkAvailable()) {
            getMsgDispatcher().dispatchFailureMsg(msg, IMSConfig.MSG_CONTENT_TYPE_MESSAGE_NETWORK_FAIL);
            return;
        }
        if (mChannel == null) {
            Logger.print("发送消息失败，channel为空");
            getMsgDispatcher().dispatchFailureMsg(msg, IMSConfig.MSG_CONTENT_TYPE_MESSAGE_SEND_FAIL);
            resetConnect();
            return;
        }
        if (isJoinTimeoutManager) {
//            getMsgTimeoutTimerManager().add(msg);
        }// 二者存其一
        getRequestTimeoutTimerManager().add(msg);
        try {
//            InetSocketAddress address = (InetSocketAddress) mChannel.localAddress();
//            Logger.print("ChannelId", mChannel.id().toString() + "[" + address.getPort() + "]");
            mChannel.writeAndFlush(msg).addListener(future -> Logger.print("Send message success"));
        } catch (Exception e) {
            Logger.print("发送消息失败，reason：" + e.getMessage() + "\tmessage=" + msg);
        }
    }

    @Override
    public void sendRequestMsg(BusinessProtocolMessageStandard.CheckUnifiedEntranceMessage msg) {
        if (msg == null) {
            Logger.print("发送消息失败，消息为空");
            return;
        }
        if (!isNetworkAvailable()) {
            getMsgDispatcher().dispatchFailureMsg(msg, IMSConfig.MSG_CONTENT_TYPE_MESSAGE_NETWORK_FAIL);
            return;
        }
        if (mChannel == null) {
            Logger.print("发送消息失败，channel为空");
            getMsgDispatcher().dispatchFailureMsg(msg, IMSConfig.MSG_CONTENT_TYPE_MESSAGE_SEND_FAIL);
            resetConnect();
            return;
        }
        getRequestTimeoutTimerManager().add(msg);
        try {
            InetSocketAddress address = (InetSocketAddress) mChannel.localAddress();
            Logger.print("ChannelId", mChannel.id().toString() + "[" + address.getPort() + "]");
            mChannel.writeAndFlush(msg).addListener(future -> Logger.print("Send message success"));
        } catch (Exception e) {
            Logger.print("发送消息失败，reason：" + e.getMessage() + "\tmessage=" + msg);
        }
    }

    @Override
    public int getReconnectInterval() {
        if (mOnEventListener != null && mOnEventListener.getReconnectInterval() > 0)
            return mOnEventListener.getReconnectInterval();
        return IMSConfig.DEFAULT_RECONNECT_INTERVAL;
    }

    @Override
    public int getConnectTimeout() {
        if (mOnEventListener != null && mOnEventListener.getConnectTimeout() > 0)
            return mOnEventListener.getConnectTimeout();
        return IMSConfig.DEFAULT_CONNECT_TIMEOUT;
    }

    @Override
    public int getResendCount() {
        if (mOnEventListener != null && mOnEventListener.getResendCount() > 0)
            return mOnEventListener.getResendCount();
        return IMSConfig.DEFAULT_RESEND_COUNT;
    }

    @Override
    public int getResendInterval() {
        if (mOnEventListener != null && mOnEventListener.getReconnectInterval() > 0)
            return mOnEventListener.getResendInterval();
        return IMSConfig.DEFAULT_RESEND_INTERVAL;
    }

    @Override
    public int getRequestTimeoutMillis() {
        if (mOnEventListener != null && mOnEventListener.getRequestTimeoutMillis() > 0)
            return mOnEventListener.getRequestTimeoutMillis();
        return IMSConfig.DEFAULT_REQUEST_TIMEOUT_MILLIS;
    }

    @Override
    public void setConnectStatusCallback(OnConnectStatusCallback callback) {
        mConnectStatusCallback = null;
        mConnectStatusCallback = callback;
    }

    @Override
    public MsgDispatcher getMsgDispatcher() {
        return mMsgDispatcher;
    }

    @Override
    public MsgTimeoutTimerManager getMsgTimeoutTimerManager() {
        return mMsgTimeoutTimerManager;
    }

    @Override
    public RequestTimeoutTimerManager getRequestTimeoutTimerManager() {
        return mRequestTimeoutTimerManager;
    }

    @Override
    public ExecutorServiceFactory getLoopGroup() {
        return mResetConnectExecutor;
    }

    // 初始化Bootstrap
    private void initBootstrap() {
        EventLoopGroup mLoopGroup = new NioEventLoopGroup(4);
        mBootstrap = new Bootstrap();
        mBootstrap.group(mLoopGroup).channel(NioSocketChannel.class);
        // 设置该选项以后，如果在两小时内没有数据的通信时，TCP会自动发送一个活动探测数据报文
        mBootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        // 设置禁用nagle算法
        mBootstrap.option(ChannelOption.TCP_NODELAY, true);
        // 设置连接超时时长
        mBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());
        // 设置初始化Channel
        mBootstrap.handler(new TCPChannelInitializer(this));
    }

    // 连接
    private void doConnect(boolean isFirst) {
        mConnectFailureRetryCount = 0;
        if (!isFirst) {
            Threader.sleep(IMSConfig.DEFAULT_RECONNECT_INTERVAL);
        }

        if (isClosed) {
//            throw new NullPointerException("NettyHttpClient is closed");
            Logger.print("NettyHttpClient is closed");
            return;
        }
        if (!isConnecting) {
            synchronized (this) {
                if (!isClosed && !isConnecting) {
                    isConnecting = true; // 表示正在连接中
                    onConnectStatusCallback(IMSConfig.CONNECT_STATE_CONNECTING);
                    closeChannel(); // 连接开始前先关闭通道
                    mResetConnectExecutor.execBossTask(new ResetConnectRunnable(isFirst));
                }
            }
        }
    }

    // 连接回调
    private void onConnectStatusCallback(int connectStatus) {
        mConnectStatus = connectStatus;
        switch (connectStatus) {
            case IMSConfig.CONNECT_STATE_CONNECTING:
                Logger.print("ims连接中...");
                if (mConnectStatusCallback != null)
                    mConnectStatusCallback.onConnecting();
                break;
            case IMSConfig.CONNECT_STATE_SUCCESSFUL:
                Logger.print(String.format("ims连接成功Server(%s)", mServerUrl));
                if (mConnectStatusCallback != null)
                    mConnectStatusCallback.onConnected();
                // 连接成功，检查消息发送超时管理器里是否有发送超时的消息，如果有，则全部重发
                getMsgTimeoutTimerManager().onResetConnected();
                break;
            case IMSConfig.CONNECT_STATE_FAILURE:
            default:
                Logger.print("ims连接失败");
                if (mConnectStatusCallback != null)
                    mConnectStatusCallback.onConnectFailed();
                break;
        }
    }

    // 关闭通道
    private void closeChannel() {
        try {
            if (mChannel != null) {
                try {
                    removeHandler(TCPReadHandler.class.getSimpleName());
                    removeHandler(IdleStateHandler.class.getSimpleName());
                } finally {
                    try {
                        mChannel.close();
                    } catch (Exception e) {
                        Logger.print(e.getMessage());
                    }
                    try {
                        mChannel.eventLoop().shutdownGracefully();
                    } catch (Exception ex) {
                        Logger.print(ex.getMessage());
                    }
                    mChannel = null;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
            Logger.print("关闭channel出错，reason:" + e.getMessage());
        }
    }

    private void removeHandler(String handlerName) {
        try {
            if (mChannel.pipeline().get(handlerName) != null) {
                mChannel.pipeline().remove(handlerName);
            }
        } catch (Exception e) {
            e.printStackTrace();
            Logger.print("移除handler失败，handlerName=" + handlerName);
        }
    }

    // 重连任务
    private class ResetConnectRunnable implements Runnable {

        private boolean isFirst;

        public ResetConnectRunnable(boolean isFirst) {
            this.isFirst = isFirst;
        }

        @Override
        public void run() {
            try {
                while (!isClosed) {
                    if (!isNetworkAvailable()) {
                        onConnectStatusCallback(IMSConfig.CONNECT_STATE_FAILURE);
                        break;
                    }
                    int status = reConnect();
                    if (status == IMSConfig.CONNECT_STATE_SUCCESSFUL) {
                        onConnectStatusCallback(status);
                        break;
                    } else if (status == IMSConfig.CONNECT_STATE_FAILURE) {
                        mConnectFailureRetryCount++;
                        if (mConnectFailureRetryCount >= IMSConfig.CONNECT_FAILURE_RETRY_COUNT) {
                            onConnectStatusCallback(status);
                            break;
                        }
                        Logger.print("连接失败，" + (IMSConfig.DEFAULT_RECONNECT_INTERVAL / 1000) + "秒后重试！");
                        Threader.sleep(IMSConfig.DEFAULT_RECONNECT_INTERVAL);
                    }
                }
            } finally {
                isConnecting = false;   // 连接任务停止
            }
        }
    }

    private int reConnect() {
        if (!isClosed) {
            try {
                if (mBootstrap != null)
                    mBootstrap.group().shutdownGracefully();
            } finally {
                mBootstrap = null;
            }
            initBootstrap();
            return connectServer();
        }
        return IMSConfig.CONNECT_STATE_FAILURE;
    }

    private int connectServer() {
        if (StringUtil.isNullOrEmpty(mServerUrl))
            return IMSConfig.CONNECT_STATE_FAILURE;

        String[] address = mServerUrl.split(":");
        if (address.length < 2)
            return IMSConfig.CONNECT_STATE_FAILURE;
        for (int i = 0; i < IMSConfig.DEFAULT_RECONNECT_COUNT; i++) {
            if (isClosed || !isNetworkAvailable())
                return IMSConfig.CONNECT_STATE_FAILURE;
            if (mConnectStatus != IMSConfig.CONNECT_STATE_CONNECTING)
                onConnectStatusCallback(IMSConfig.CONNECT_STATE_CONNECTING);
            Logger.print(String.format("正在进行『%s』的第『%s』次连接，当前重连延时时长为『%sms』", mServerUrl, String.valueOf(i), String.valueOf(i * getReconnectInterval())));
            try {
                mCurrentHost = address[0];
                try {
                    mCurrentPort = Integer.parseInt(address[1]);
                } catch (NumberFormatException e) {
                    return IMSConfig.CONNECT_STATE_FAILURE;
                }
                toServer();
                if (mChannel != null) {
                    // channel不为空，即认为连接已成功
                    Logger.print(String.format("连接Server(%s)成功", mServerUrl));
                    return IMSConfig.CONNECT_STATE_SUCCESSFUL;
                } else {
                    // 连接失败，则线程休眠n * 重连间隔时长
                    Thread.sleep(i * getReconnectInterval());
                }
            } catch (InterruptedException e) {
                close();
                break;  //线程中断，强制关闭
            }
        }
        return IMSConfig.CONNECT_STATE_FAILURE;
    }

    private void toServer() {
        try {
            mChannel = mBootstrap.connect(mCurrentHost, mCurrentPort).sync().channel();
        } catch (Exception e) {
            Threader.sleep(500);
            Logger.print(String.format("连接Server(%s)失败", mServerUrl) + e.getLocalizedMessage());
            mChannel = null;
        }
    }
}
